#ifndef __KAFKA_METRIC_REPORTER_HPP__
#define __KAFKA_METRIC_REPORTER_HPP__

#include <string>
#include <memory>

#include <librdkafka/rdkafkacpp.h>

#include "metrics/MetricReporter.hpp"

namespace metrics
{
    class JsonMetricFormatter;

    class DefaultKafkaDRCb : public RdKafka::DeliveryReportCb
    {
    public:
        DefaultKafkaDRCb() = default;
        void dr_cb(RdKafka::Message &message) override;
    };

    class DefaultKafkaEventCb : public RdKafka::EventCb
    {
    public:
        DefaultKafkaEventCb() = default;
        void event_cb(RdKafka::Event &event) override;
    };

    class KafkaMetricReporter : public MetricReporter
    {
    private:
        RdKafka::Conf *_conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
        JsonMetricFormatter *_formatter;
        std::string _brokers; //kafka _brokers
        std::string _topic;   //kafka topic
        std::unique_ptr<RdKafka::EventCb> _event_cb;
        std::unique_ptr<RdKafka::DeliveryReportCb> _dr_cb;
        RdKafka::Producer *_producer;

    public:
        KafkaMetricReporter(const KafkaMetricReporter &) = delete;
        KafkaMetricReporter &operator=(const KafkaMetricReporter &) = delete;
        KafkaMetricReporter(const std::string &brokers, const std::string &topic,
                            std::unique_ptr<RdKafka::EventCb> event_cb = std::unique_ptr<DefaultKafkaEventCb>(new DefaultKafkaEventCb()),
                            std::unique_ptr<RdKafka::DeliveryReportCb> dr_cb = std::unique_ptr<DefaultKafkaDRCb>(new DefaultKafkaDRCb()));
        ~KafkaMetricReporter();

    protected:
        virtual void run();
        virtual void extra_metrics();
    };

} // namespace metrics

#endif // __KAFKA_METRIC_REPORTER_HPP__